/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */
package org.singularity.io.util;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.zip.GZIPOutputStream;

import org.singularity.io.CloseablePipe;

/**
 * The utility to serialize a values as a sequence of tuples 
 * with fixed arity to the binary file. This utility gains 
 * benefit in case of tuples are ordered i.e. there is a good 
 * chance to have data represented compactly.
 * 
 * @author vjache
 * 
 * @see TupleDeserializerIterator
 *
 */
public abstract class TupleSerializerPipe<T> implements CloseablePipe<T> {

	protected final DataOutputStream _out;
	private final int _tupleSize;
	
	private Object[] _previousTuple;
	private ArrayList<T> _currentTuple;
	
	/**
	 * @throws IOException 
	 * 
	 */
	public TupleSerializerPipe(File aFile,int aTupleSize,boolean aZipped) throws IOException {
		_tupleSize = aTupleSize;
		_currentTuple = new ArrayList<T>(_tupleSize);
		_previousTuple = new Object[_tupleSize];
		for(int i=0;i<_tupleSize;i++)
			_previousTuple[i] = new Object();
		final FileOutputStream fos = new FileOutputStream(aFile);
		fos.write(aZipped?1:0);
		_out = new DataOutputStream(
				new BufferedOutputStream(
						aZipped?
						new GZIPOutputStream(
								fos)
						:fos,
						64*1024));
		_out.writeInt(_tupleSize);
	}

	/* (non-Javadoc)
	 * @see java.io.Closeable#close()
	 */
	@Override
	public void close() throws IOException {
		flush();
		_out.close();
	}

	/* (non-Javadoc)
	 * @see org.singularity.io.Pipe#flush()
	 */
	@Override
	public void flush() throws IOException {}

	/* (non-Javadoc)
	 * @see org.singularity.io.Pipe#push(java.lang.Object)
	 */
	@Override
	public void push(T aItem) throws IOException {
		_currentTuple.add(aItem);
		if(_currentTuple.size()==_tupleSize)
		{
			int startPos = 0;
			while(startPos<_tupleSize && _currentTuple.get(startPos).
					equals(_previousTuple[startPos]))
				startPos++;
			
			_out.writeByte(startPos);
			for(int i = startPos;i<_tupleSize;i++)
			{
				serialize(_currentTuple.get(i));
			}
			
			_previousTuple = _currentTuple.toArray();
			_currentTuple.clear();
		}
	}

	/**
	 * Serializes an item to internal data output stream.
	 * @param aItem
	 * @throws IOException
	 */
	protected abstract void serialize(T aItem) throws IOException ;

}
